package pers.wk.queues.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import pers.wk.config.MQConfig;
import pers.wk.utils.MQConnectionUtils;

/**
 * 消息发布确认机制-单条确认
 */
@Slf4j
public class ConfirmProducer {

    public static void main(String[] args) {
        //1.ConnectionFactory
        ConnectionFactory connectionFactory = MQConnectionUtils.getConnectionFactory("/");
        Connection connection = null;
        Channel channel = null;
        try {
            //2.创建连接
            connection = connectionFactory.newConnection();
            //3.通过连接获取通道Channel
            channel = connection.createChannel();
            //创建交换机
            channel.exchangeDeclare(MQConfig.ExchangeName.DIRECT_EXCHANGE, MQConfig.ExchangeType.direct);
            //创建队列
            channel.queueDeclare(MQConfig.QueueNames.EMAIL_QUEUE,
                    false, false, false, null);
            //队列绑定交换机
            String routingKey = "email";
            channel.queueBind(MQConfig.QueueNames.EMAIL_QUEUE, MQConfig.ExchangeName.DIRECT_EXCHANGE, routingKey);

            System.out.println("路由模式模式direct-生产者启动。。。");

            // 启用发布者确认模式 只能确定是否投递到了交换机，不能确认交换机投递到指定的队列中
            channel.confirmSelect();

            for (int i = 0; i < 10; i++) {
                String message = "hello rabbitmq " + i;
                channel.basicPublish(MQConfig.ExchangeName.DIRECT_EXCHANGE, routingKey, null, message.getBytes());
                // 一条一条确认，返回为true，则表示发送成功
                boolean waitForConfirms = channel.waitForConfirms();
                System.out.println("waitForConfirms = " + waitForConfirms);
                if (waitForConfirms) {
                    log.info("send success, routingKey: {}, message: {}", routingKey, message);
                } else {
                    log.info("send fail, routingKey: {}, message: {}", routingKey, message);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            MQConnectionUtils.close(connection, channel);
        }
    }
}
